/*
 * ===================================================================
 *  
 *      Filename: log_srv_protocol_tcp.c
 *
 *   Description: This is the tcp plugin,we used the EPOLL to improve 
 *                the performace of the hplogd.the encapulation packets
 *                are welcomed(pack small packets into a large packet)
 *
 *       Version:  1.0
 *        Create:  2010/07/13
 *
 *        Author:  Keven Zhu     kevenzhu@tencent.com
 *     Copyright:  Reserved by Tencent.com Co.ltd
 *   
 * ====================================================================
 */
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <netdb.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <netinet/in.h>
#include <signal.h>
#include <linux/unistd.h>
#include <unistd.h>
#include "lib/list.h"
#include "lib/util.h"

#define  __USE_GNU
#include <sched.h>
#include <pthread.h>

#define WORKING_THREAD         1
#define WORKING_THREAD_MASK    3
#define MAX_TCP_BACKLOG        2048
#define MAX_EVENT_NUM          2048
#define MAX_EPOLL_SIZE         2048

thread_info_t tcp_thread_pool;
pthread_t     epoll_pid;

/*global buffer for receiving the tcp socket*/
char *tcp_global_buff = NULL;

extern long recv_counter;
extern long recv_traffic;

/******************************************************************/
/*for tcp hash buff*/
struct tcp_buffer_hash_list
{
    struct hlist_head    head;
    pthread_spinlock_t   lock;
};

struct tcp_buffer_hash_node
{
    struct hlist_node   list;
    int                 sockfd;
    int                 offset;
    int                 msg_len;
    char                buff[MAX_LOG_SIZE];
};

/* we choose hlist for the client info, and we assume that
 * most of the client is located in the same network
 */
#define HASH_MASK_TCP_BUFFER  1023
struct  tcp_buffer_hash_list tcp_remote_client_list[1024];

/*we only support the ipv4 clinet*/
int add_tcp_buffer_node(int sockfd,char* msg, int msg_len)
{
    struct hlist_node *loop_node;
    struct tcp_buffer_hash_node *tmp;
    struct tcp_buffer_hash_node *loop_container;
    struct tcp_buffer_hash_list *tmp_head = &tcp_remote_client_list[sockfd&HASH_MASK_TCP_BUFFER];

    pthread_spin_lock(&tmp_head->lock);
    /*it's the first node*/
    if(hlist_empty(&tmp_head->head))
    {
        tmp = (struct tcp_buffer_hash_node *)malloc(sizeof(struct tcp_buffer_hash_node));
        if(tmp)
        {
            tmp->sockfd = sockfd;
            //tmp->offset = offset;
            tmp->msg_len= msg_len;
            memcpy(tmp->buff,msg,msg_len);
            hlist_add_head(&tmp->list,&tmp_head->head);
        }
        else
        {
            printf("No memory for the hlist node, exit\n");
            tmp = NULL;
        }
    }
    /*to test confict or exist one*/
    else
    {
        hlist_for_each_entry(loop_container,loop_node,&tmp_head->head,list)
        {
            if(loop_container->sockfd == sockfd)
            {
                pthread_spin_unlock(&tmp_head->lock);
                printf("failed to alloc a new node, the old one is exist\n");
                return FALSE;
            }
        }
        tmp = (struct tcp_buffer_hash_node *)malloc(sizeof(struct tcp_buffer_hash_node));
        tmp->sockfd = sockfd;
        //tmp->offset = offset;
        tmp->msg_len= msg_len;
        memcpy(tmp->buff,msg,msg_len);
        hlist_add_head(&tmp->list,&tmp_head->head);
    }
    pthread_spin_unlock(&tmp_head->lock);
    return OK;
}

/*add an client into the list*/
int remove_tcp_buffer_node(int sockfd)
{
    struct hlist_node *loop_node,  *n_node;
    struct tcp_buffer_hash_node *loop_container;
    struct tcp_buffer_hash_list *tmp_head = &tcp_remote_client_list[sockfd&HASH_MASK_TCP_BUFFER];

    pthread_spin_lock(&tmp_head->lock);
    hlist_for_each_entry_safe(loop_container,loop_node,n_node,&tmp_head->head,list)
    {
        if(loop_container->sockfd == sockfd)
        {
            hlist_del(&loop_container->list);
            pthread_spin_unlock(&tmp_head->lock);
            free((char*)loop_container);
            return OK;
        }
    }
    pthread_spin_unlock(&tmp_head->lock);
    return FALSE;
}
/*we only support the ipv4 clinet*/
struct tcp_buffer_hash_node* get_tcp_buffer_node(int sockfd)
{
    struct hlist_node *loop_node;
    struct tcp_buffer_hash_node *tmp = NULL;
    struct tcp_buffer_hash_node *loop_container;
    struct tcp_buffer_hash_list *tmp_head = &tcp_remote_client_list[sockfd&HASH_MASK_TCP_BUFFER];

    pthread_spin_lock(&tmp_head->lock);
    hlist_for_each_entry(loop_container,loop_node,&tmp_head->head,list)
    {
        if(loop_container->sockfd == sockfd)
        {
            tmp = loop_container;
            break;
        }
    }
    pthread_spin_unlock(&tmp_head->lock);
    return tmp;
}


/******************************************************************/



static inline void *tcp_memcpy(void *to, const void *from, size_t n)
{
        int d0, d1, d2;
        asm volatile("rep ; movsl\n\t"
             "movl %4,%%ecx\n\t"
             "andl $3,%%ecx\n\t"
             "jz 1f\n\t"
             "rep ; movsb\n\t"
             "1:"
             : "=&c" (d0), "=&D" (d1), "=&S" (d2)
             : "0" (n / 4), "g" (n), "1" ((long)to), "2" ((long)from)
             : "memory");
    return to;
}

int tcp_buffer_enqueue(char *buff)
{
    int   ipos = 0;
    char* send_buff = NULL;
    /*for the tcp, we only have one queue*/
    thread_info_t *thread = &tcp_thread_pool;
    
    pthread_mutex_lock(&counter_mutex);
    recv_counter ++;
    recv_traffic += *(short*)buff;
    pthread_mutex_unlock(&counter_mutex);
    
    ipos = enqueue(thread,buff);
    if(NORMAL_THRESHOLD == ipos)
    {
        //normal case, don't send the traffic
        return OK;
    }
    else if(HIGH_THRESHOLD == ipos)
    {
        //little space left,try to output
    }
    else if(URGENCY_THRESHOLD == ipos)
    {
        /*wait until the packet can be handled*/
        //while(URGENCY_THRESHOLD == enqueue(thread,buff));
        /*how to reduce the change of drop*/
        free(buff);
             
        printf("queue is full,send the packet anyway\n");
    }
    else
    {
        //not care until now
        printf("I don't know what's going on, drop the packet\n");
        free(buff);
        return FALSE;
    }

    while((send_buff = (char*)dequeue(thread)))
    {
        #ifdef DEBUG_LOG
        printf("recv buff2 %p\n",send_buff);
        #endif
        /*we process the packet suddenly*/
        main_framer_process_pkt(send_buff);
    }
    return OK;
}

int manipulate_tcp_msg(int sockfd,char *msg,int total_msg_size)
{    
    struct tcp_buffer_hash_node *tcp_hash_node ;
    char   *buff,*tmp_buff;
    int    iret,pkt_len,msg_size;
    
    /*hold the buffer*/
    tmp_buff = msg;
    msg_size = total_msg_size;

    /*construct the packet from the msg buffer*/
    tcp_hash_node =  get_tcp_buffer_node(sockfd);
    if(tcp_hash_node)
    {
        buff = malloc(MAX_LOG_SIZE);
        if(!buff)
        {
            printf("failed to get the memory, we don't have the length of the packet");
            memcpy(tcp_hash_node->buff + tcp_hash_node->msg_len,tmp_buff,4);
            pkt_len = *((short *)tcp_hash_node->buff);
            iret = tcp_hash_node->msg_len; //last partition
            remove_tcp_buffer_node(sockfd);
        }
        else
        {

            /*copy the remaining to the buffer*/
            memcpy(buff,tcp_hash_node->buff,tcp_hash_node->msg_len);
            /*copy the startup of the new packet to the buffer*/
            memcpy(buff+tcp_hash_node->msg_len,msg,4);//at least we can know the length of the msg
            pkt_len = *((short *)buff); //get the message size
            iret    = tcp_hash_node->msg_len;
            memcpy(buff+iret,msg,pkt_len - iret);//copy the remaining one
            if(msg_size+iret <pkt_len) //the incoming packet is short
            {
                memcpy(tcp_hash_node->buff+iret , msg, msg_size);
                tcp_hash_node->msg_len = tcp_hash_node->msg_len + msg_size;
                free(buff);//release the new buffer
                return OK;
            }
            else
            {
                remove_tcp_buffer_node(sockfd);
                tcp_buffer_enqueue(buff);
            }
        }
        /*now we got a new packet, and shift the point to the start point*/
        msg_size = msg_size - pkt_len + iret;
        tmp_buff = tmp_buff + pkt_len - iret;

        if(msg_size == 0) //the buffer has been processed
            return OK;
    }
    else
    {
        //printf("no node in the hash\n");
    }

    /*loop to read the log message from the socket msg*/
    pkt_len = *((short *)tmp_buff);
    if(pkt_len > MAX_LOG_SIZE)
    {
        return FALSE;
    }

    while((pkt_len <= msg_size) && msg_size)
    {
        buff = malloc(pkt_len);
        if(!buff)
        {
            printf("no new memory for the packet\n");
        }
        else
        {
            memcpy(buff,tmp_buff,pkt_len);
            tcp_buffer_enqueue(buff);
        }
        tmp_buff = tmp_buff + pkt_len;
        msg_size = msg_size - pkt_len;
        if(msg_size>3)
            pkt_len  = *((short *)tmp_buff);
        else
            pkt_len  = 99999999; //impossible value
    }

    /*check if there is partition packet for the next time*/
    if(msg_size)
    {
        add_tcp_buffer_node(sockfd,tmp_buff, msg_size); 
    }

    return OK;
}

int tcp_process_log(int sockfd)
{
    int  recv_size=0;

    recv_size = recv(sockfd,tcp_global_buff,MAX_LOG_SIZE ,0);
    if(recv_size>0)
    {
            
        #ifdef DEBUG_LOG
        printf("buff1 %p\n ",buff);
        for(ipos=0;ipos<recv_size;ipos++)
        {
            printf("%02x",buff[ipos]);
        }
        printf("\n");
        #endif

        return manipulate_tcp_msg(sockfd,tcp_global_buff,recv_size);
    }
    else
    {
        remove_tcp_buffer_node(sockfd);
        /*remove the client from the client list*/
        //remove_remote_client();
        if(-1 == recv_size)
        {
            close(sockfd);
            if(EAGAIN == errno)
            {
                #ifdef DEBUG_LOG
                printf("recv time out, try to flush the buffer\n");
                #endif
                return FALSE;
            }
            else
                return FALSE;
        }
        else
        {
            /*the client has been closed*/
            close(sockfd);
            return FALSE; 
        }
    }
}

void* epoll_routine(void *argv)
{
    unsigned int ipos,i;
    struct timeval tv;
    cpu_set_t mask;
    char *send_buff;
    int listen_tcp_sockfd,epfd,nfds,client_addrlen,client_fd;
    struct sockaddr_in serv_addr,client_addr;
    struct epoll_event ev,events[MAX_EVENT_NUM];

    /*the tcp deamon was bind to the CPU0*/
    CPU_ZERO(&mask);
    CPU_SET(3, &mask);
    int ret = pthread_setaffinity_np(pthread_self(), sizeof(mask), &mask);
    if (ret < 0)
        perror("sched_setaffinity");

    /*init the queue*/
    init_queue(&tcp_thread_pool,MAX_WORK_QUEUE);

    listen_tcp_sockfd = socket(AF_INET,SOCK_STREAM,0);
    if(listen_tcp_sockfd == -1)
    {
        perror("error to create the tcp socket");
        exit(1);
    }

    /*enlarge the socket buffer*/
    ipos = 4 * 1024 *1024;
    setsockopt(listen_tcp_sockfd,SOL_SOCKET,SO_RCVBUF,(char*)&ipos,sizeof(ipos));

    /*set the address to reuse*/
    ipos = TRUE;
    setsockopt(listen_tcp_sockfd,SOL_SOCKET,SO_REUSEADDR,&ipos,sizeof(ipos));

    /*set the recv timeout value*/
    tv.tv_sec = 3;
    tv.tv_usec= 0;
    setsockopt(listen_tcp_sockfd,SOL_SOCKET,SO_RCVTIMEO,&tv,sizeof(tv));   

    serv_addr.sin_family = AF_INET;
    serv_addr.sin_port   = htons(SPORT);
    serv_addr.sin_addr.s_addr   = htonl(INADDR_ANY);//*((struct in_addr*)host->h_addr);
    bzero(&(serv_addr.sin_zero),8);

    if(-1 ==bind(listen_tcp_sockfd,(struct sockaddr*)&serv_addr,sizeof(struct sockaddr)))
    {
        perror("error to bind the tcp connection");
        close(listen_tcp_sockfd);
        exit(1);
    };

    if(-1 == listen(listen_tcp_sockfd,MAX_TCP_BACKLOG))
    {
        perror("error to listen the tcp socket");
        close(listen_tcp_sockfd);
        exit(1);
    }

    if(-1 == (epfd = epoll_create(MAX_EPOLL_SIZE)))
    {
        perror("error to create the epoll");
        close(listen_tcp_sockfd);
        exit(1);
    }

    ev.data.fd = listen_tcp_sockfd;
    ev.events  = EPOLLIN ; /*for surence,we use level triggle*/
    //ev.events  = EPOLLIN | EPOLLLT; /*for surence,we use level triggle*/
    if(-1 == epoll_ctl(epfd,EPOLL_CTL_ADD,listen_tcp_sockfd,&ev))
    {
        perror("error to add the listen socket to the epoll");
        close(epfd);
        close(listen_tcp_sockfd);
        exit(1);
    } 

    while(1)
    {
        nfds = epoll_wait(epfd,events,MAX_EVENT_NUM,3000);
        if(nfds>0)
        {
            /*socket available*/

        }
        else if(!nfds)
        {
            /*time out, flush the queue*/
            while((send_buff = (char*)dequeue(&tcp_thread_pool)))
            {
                #ifdef DEBUG_LOG
                printf("time out buff2 %p\n",send_buff);
                #endif
                /*we process the packet suddenly*/
                main_framer_process_pkt(send_buff);
            }
            continue;
        }
        else
        {
            /*error occurs*/
            perror("error in epoll");
            continue;
        }
        /*we try to read the data from the socket*/
        for(i=0;i<nfds;i++)
        {
            if(events[i].data.fd == listen_tcp_sockfd)
            {
                /*a new client call in*/
                client_addrlen = sizeof(struct sockaddr_in);
                client_fd = accept(listen_tcp_sockfd,(struct sockaddr*)&client_addr,(socklen_t*)&client_addrlen);
                if(-1 == client_fd)
                {
                    perror("error in accept the new client");
                    continue;
                }
                /*remove the partial hash*/
                remove_tcp_buffer_node(client_fd);
                /*register the client address to the policy modules*/
                add_remote_client(client_addr.sin_addr.s_addr);
                
                /*set the keepalive of the client socket*/
                ipos = 1;
                if(setsockopt(client_fd, SOL_SOCKET, SO_KEEPALIVE, &ipos, sizeof(ipos)) < 0) 
                {
                    perror("setsockopt()");
                }

                ev.data.fd = client_fd;
                ev.events  = EPOLLIN ;
                //ev.events  = EPOLLIN |EPOLLLT;
                if(-1 == epoll_ctl(epfd,EPOLL_CTL_ADD,client_fd,&ev))
                {
                    perror("error to add the new client to the epoll");
                    close(client_fd);
                    continue;
                }
            }
            else
            {
                /*the socket is available for reading new packet*/
                tcp_process_log(events[i].data.fd);
            }
        }
    }

    exit(1);
}

int init_protocol_tcp()
{
    int iret = 0;
    struct  tcp_buffer_hash_list *tmp_head = (struct tcp_buffer_hash_list *)&tcp_remote_client_list;

    tcp_global_buff = (char *)malloc(MAX_LOG_SIZE); /*get the free packet*/
    if(!tcp_global_buff)
    {
        perror("failed to alloc the tcp global memory");
        return FALSE;
    }
    
    memset(&tcp_remote_client_list,0,sizeof(struct tcp_buffer_hash_list) * 1024);
    
    for(iret =0;iret<1024;iret++)
    {
        INIT_HLIST_HEAD(&tmp_head->head);
        if(pthread_spin_init(&tmp_head->lock,PTHREAD_PROCESS_PRIVATE))
        {
            perror("failed to initialize the spin lock");
            return FALSE;
        }
        tmp_head++;
    }
    
    iret = pthread_create(&epoll_pid,NULL,epoll_routine,NULL);
    if(iret)
    {
        free(tcp_global_buff);
        return FALSE;
    }
    else
    {
        return OK;
    }
}

int protocol_tcp_join()
{
    pthread_join(epoll_pid,NULL);
    return OK;
}

int finit_protocol_tcp()
{
    int i = 0;

    printf("    stop the tcp epoll thread\n");
    pthread_cancel(epoll_pid);

    printf("    free the global tcp buffer\n");
    free(tcp_global_buff);

    /*destroy the queue*/
    printf("    destroy the pthread private queue NO.%d\n",i);
    finit_queue((thread_info_t *)&tcp_thread_pool);
    printf("    success to unload the tcp protocol modules\n");
    return OK;
}

